package flink.api_study.state;

import flink.api_study.source.Person;
import flink.api_study.source.PersonSource;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.sql.Timestamp;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.List;

/**
 * Created by tzs on 2022/8/2.
 */
public class StateTest {
    public static void main(String[] args) throws Exception{
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);

        DataStream<Person> stream = env.addSource(new PersonSource());

        stream.map(new MyMapFuction1()).print();
        env.execute();
    }

    static class MyMapFuction1 implements MapFunction<Person,Integer>{
        private Integer count = 0;

        @Override
        public Integer map(Person o) throws Exception {
            count++;
            return count;
        }
    }









    static class MyMapFuction2 implements MapFunction<Person,Integer>,ListCheckpointed<Integer> {
        private Integer count = 0;

        @Override
        public Integer map(Person o) throws Exception {
            count++;
            return count;
        }


        @Override
        public List<Integer> snapshotState(long checkpointId, long timestamp) throws Exception {
            return Collections.singletonList(count);
        }

        @Override
        public void restoreState(List<Integer> state) throws Exception {
            for (Integer num:state) {
                count += num;
            }
        }
    }
}
